package com.wande.dataplatform.etl.node;

import cn.hutool.core.util.StrUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * 转换节点 - 数据转换和清洗
 *
 * @author dataplatform
 */
@Slf4j
@Component
public class TransformNode implements EtlNode {

    @Override
    public EtlNodeResult execute(EtlContext context) {
        try {
            List<Map<String, Object>> inputData = context.getDataSet();
            if (inputData == null || inputData.isEmpty()) {
                return EtlNodeResult.success(new ArrayList<>());
            }
            
            Map<String, Object> config = context.getConfig();
            
            // 获取转换配置
            @SuppressWarnings("unchecked")
            List<Map<String, Object>> fieldMappings = (List<Map<String, Object>>) config.get("fieldMappings");
            @SuppressWarnings("unchecked")
            List<Map<String, Object>> filters = (List<Map<String, Object>>) config.get("filters");
            
            List<Map<String, Object>> transformedData = new ArrayList<>();
            long successCount = 0;
            long failedCount = 0;
            
            for (Map<String, Object> row : inputData) {
                try {
                    // 应用过滤器
                    if (filters != null && !applyFilters(row, filters)) {
                        continue;
                    }
                    
                    // 应用字段映射
                    Map<String, Object> transformedRow = new HashMap<>();
                    if (fieldMappings != null) {
                        for (Map<String, Object> mapping : fieldMappings) {
                            String sourceField = (String) mapping.get("sourceField");
                            String targetField = (String) mapping.get("targetField");
                            String transformType = (String) mapping.get("transformType");
                            
                            Object value = row.get(sourceField);
                            Object transformedValue = applyTransform(value, transformType);
                            transformedRow.put(targetField, transformedValue);
                        }
                    } else {
                        // 如果没有配置映射，直接复制所有字段
                        transformedRow.putAll(row);
                    }
                    
                    transformedData.add(transformedRow);
                    successCount++;
                    
                } catch (Exception e) {
                    log.warn("转换行数据失败: {}", e.getMessage());
                    failedCount++;
                }
            }
            
            log.info("转换节点处理完成: 成功 {} 行, 失败 {} 行", successCount, failedCount);
            context.getStatistics().addSuccessRows(successCount);
            context.getStatistics().addFailedRows(failedCount);
            
            EtlNodeResult result = EtlNodeResult.success(transformedData);
            result.setSuccessRows(successCount);
            result.setFailedRows(failedCount);
            return result;
            
        } catch (Exception e) {
            log.error("转换节点执行失败", e);
            return EtlNodeResult.failure("转换节点执行失败: " + e.getMessage());
        }
    }

    /**
     * 应用过滤器
     */
    private boolean applyFilters(Map<String, Object> row, List<Map<String, Object>> filters) {
        for (Map<String, Object> filter : filters) {
            String field = (String) filter.get("field");
            String operator = (String) filter.get("operator");
            Object value = filter.get("value");
            
            Object fieldValue = row.get(field);
            
            // 简单的过滤逻辑
            if ("equals".equals(operator)) {
                if (fieldValue == null || !fieldValue.equals(value)) {
                    return false;
                }
            } else if ("notNull".equals(operator)) {
                if (fieldValue == null) {
                    return false;
                }
            }
        }
        return true;
    }

    /**
     * 应用转换
     */
    private Object applyTransform(Object value, String transformType) {
        if (value == null || StrUtil.isBlank(transformType)) {
            return value;
        }
        
        switch (transformType) {
            case "toUpperCase":
                return value.toString().toUpperCase();
            case "toLowerCase":
                return value.toString().toLowerCase();
            case "trim":
                return value.toString().trim();
            default:
                return value;
        }
    }

    @Override
    public String getNodeType() {
        return TRANSFORM;
    }

    @Override
    public boolean validate(Map<String, Object> config) {
        return true; // 转换节点配置是可选的
    }
}
